package com.hx78.org;

import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class SubscribeAsyncExample {
  private static final ReentrantLock LOCK = new ReentrantLock();
  private static final Condition STOP = LOCK.newCondition();

  private static volatile long now = System.currentTimeMillis();

  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.


    String projectId = "mkp-dev-np1";
//    String subscriptionId = "test_by_linming_order_sub";
//    String subscriptionId = "online_product_sync.glb_dev00.online_product-sub";
    String subscriptionId = "bbbbb";

//    String projectId = "mkp-pci-e2e";
//    String subscriptionId = "online_product_sync.glb_tst03.online_product-sub";

//    String projectId = "pubsub-dev-308914";
//    String subscriptionId = "ifr-api-monitor-inv-status";
//    new Thread(new Runnable() {
//      @Override
//      public void run() {
//        subscribeAsyncExample(projectId, subscriptionId);
//      }
//    }).start();
//
//    new Thread(new Runnable() {
//      @Override
//      public void run() {
//        subscribeAsyncExample(projectId, subscriptionId);
//      }
//    }).start();
//
//    new Thread(new Runnable() {
//      @Override
//      public void run() {
//        subscribeAsyncExample(projectId, subscriptionId);
//      }
//    }).start();

    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);
//    subscribeAsyncExample(projectId, subscriptionId);

    //主线程阻塞等待，守护线程释放锁后退出
    try {
      LOCK.lock();
      STOP.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      LOCK.unlock();
    }
  }

  public static void subscribeAsyncExample(String projectId, String subscriptionId) {
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    // Instantiate an asynchronous message receiver.
    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          // Handle incoming message, then ack the received message.
//          if(System.currentTimeMillis() - now > 50000) {
//            now = System.currentTimeMillis();
//            System.out.println(new Date(now) + " Id: " + message.getMessageId() + " thread:" + Thread.currentThread().getId());
//          }

          System.out.println(new Date(now) + " Id: " + message.getMessageId() + " thread:" + Thread.currentThread().getId());

          if(message != null && message.getData() != null) {
            String s = message.getData().toStringUtf8();
//            if(s.contains("MP671459") || s.contains("10001537") || s.contains("00075577404025") || s.contains("10191470")) {
//              System.out.println("Id:" + message.getMessageId() + " data:" + s);
//            }

            if(s.contains("6090895913638346752")) {
              System.out.println("Id:" + message.getMessageId() + " data:" + s);
            }

            if(s.contains("\"op\":\"r\"")) {
              System.out.println("===ack=====");
              consumer.ack();
            }

//            if(s.contains("10667500") || s.contains("10001537")) {
//              System.out.println("data:" + s);
//            }
          }
          //System.out.println("Data: " + message.getData().toStringUtf8());
          consumer.nack();
        };

    Subscriber subscriber = null;
    subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
    // Start the subscriber.
    subscriber.startAsync().awaitRunning();
    System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
//    try {
//      subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
//      // Start the subscriber.
//      subscriber.startAsync().awaitRunning();
//      System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
//      // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
//      //subscriber.awaitTerminated(30, TimeUnit.SECONDS);
//    } catch (TimeoutException timeoutException) {
//      // Shut down the subscriber after 30s. Stop receiving messages.
//      subscriber.stopAsync();
//    }
  }
}